//! 存储层与响应回写驱动器集成演示
//! 
//! 展示存储层操作完成后如何自动回写响应的完整流程

use std::sync::{Arc, atomic::{AtomicU64, Ordering}};
use std::time::{Duration, Instant};
use std::thread;

use redis_rs2::event::{
    ResponseWritebackDriver, 
    ResponseEvent,
    NetworkCallback,
    ConnectionId,
    RequestId,
};
use redis_rs2::storage::{
    SingleThreadStorage, 
    SingleThreadCommandDispatcher,
    CommandResult,
    SetCommand,
    GetCommand,
    Command,
};

/// 模拟网络连接管理器
struct MockNetworkManager {
    active_connections: Arc<AtomicU64>,
}

impl MockNetworkManager {
    fn new() -> Self {
        Self {
            active_connections: Arc::new(AtomicU64::new(0)),
        }
    }

    /// 创建网络回调函数
    fn create_callback(&self) -> NetworkCallback {
        let connections = Arc::clone(&self.active_connections);
        
        Arc::new(move |conn_id, req_id, response_data| {
            // 模拟网络回写逻辑
            println!(
                "🌐 [网络层] 向连接 {} 发送响应 (请求 {}): {}",
                conn_id, req_id, response_data.trim()
            );
            
            // 统计活跃连接
            connections.store(conn_id.max(connections.load(Ordering::Relaxed)), Ordering::Relaxed);
        })
    }

    fn get_max_connection_id(&self) -> u64 {
        self.active_connections.load(Ordering::Relaxed)
    }
}

/// 增强的存储处理器，集成响应回写
struct EnhancedStorageProcessor {
    storage: SingleThreadStorage,
    dispatcher: SingleThreadCommandDispatcher,
    response_driver: ResponseWritebackDriver,
}

impl EnhancedStorageProcessor {
    fn new(response_driver: ResponseWritebackDriver) -> Self {
        let mut dispatcher = SingleThreadCommandDispatcher::new();
        dispatcher.start(); // 启动分发器
        
        Self {
            storage: SingleThreadStorage::new(),
            dispatcher,
            response_driver,
        }
    }

    /// 处理命令并自动回写响应
    fn process_command_with_response(
        &mut self,
        connection_id: ConnectionId,
        request_id: RequestId,
        command: Box<dyn Command>,
    ) -> Result<(), String> {
        let start_time = current_timestamp_millis();
        let command_name = command.name().to_string();
        
        println!(
            "🏪 [存储层] 处理命令: {} (连接:{} 请求:{})",
            command_name, connection_id, request_id
        );
        
        // 分发命令
        if let Err(e) = self.dispatcher.dispatch_command(command) {
            return Err(format!("命令分发失败: {}", e));
        }
        
        // 等待命令处理完成 (简化模拟)
        thread::sleep(Duration::from_millis(10));
        
        // 模拟命令执行结果（实际应由事件处理器提供）
        let result = match command_name.as_str() {
            "SET" => CommandResult::Ok,
            "GET" => CommandResult::String("模拟值".to_string()),
            _ => CommandResult::String("默认响应".to_string()),
        };
        
        // 提交响应到回写驱动器
        self.response_driver.submit_command_result(
            connection_id,
            request_id,
            result,
            command_name,
            start_time,
        )?;
        
        Ok(())
    }

    fn get_storage_stats(&self) -> String {
        let stats = self.dispatcher.get_stats();
        format!(
            "存储统计: 已分发 {} 条命令，成功率 {:.1}%，平均延迟 {}μs",
            stats.total_dispatched,
            stats.success_rate() * 100.0,
            stats.handler_stats.avg_processing_time_us
        )
    }
}

fn main() {
    println!("🎯 Redis-rs2 存储层与响应回写集成演示");
    println!("{}", "=".repeat(60));
    
    // 1. 创建网络管理器和回调
    println!("\n🌐 1. 创建网络管理器");
    let network_manager = MockNetworkManager::new();
    let network_callback = network_manager.create_callback();
    
    // 2. 创建并配置响应回写驱动器
    println!("\n📡 2. 创建响应回写驱动器");
    let mut response_driver = ResponseWritebackDriver::new("storage-integration", 5000);
    response_driver.set_network_callback(network_callback);
    
    if let Err(e) = response_driver.start() {
        eprintln!("❌ 启动响应驱动器失败: {}", e);
        return;
    }
    
    // 3. 创建增强的存储处理器
    println!("\n💾 3. 创建增强存储处理器");
    let mut storage_processor = EnhancedStorageProcessor::new(response_driver);
    
    // 4. 模拟客户端请求处理
    println!("\n👥 4. 模拟客户端请求处理");
    
    let test_commands = vec![
        // 基本 SET/GET 操作
        ("SET", 1, 101, vec!["key1".to_string(), "value1".to_string()]),
        ("GET", 1, 102, vec!["key1".to_string()]),
        ("SET", 2, 201, vec!["key2".to_string(), "value2".to_string()]),
        ("GET", 2, 202, vec!["key2".to_string()]),
        
        // 不存在的键
        ("GET", 1, 103, vec!["nonexistent".to_string()]),
        
        // 批量操作
        ("SET", 3, 301, vec!["batch1".to_string(), "data1".to_string()]),
        ("SET", 3, 302, vec!["batch2".to_string(), "data2".to_string()]),
        ("SET", 3, 303, vec!["batch3".to_string(), "data3".to_string()]),
        ("GET", 3, 304, vec!["batch1".to_string()]),
        ("GET", 3, 305, vec!["batch2".to_string()]),
        ("GET", 3, 306, vec!["batch3".to_string()]),
    ];
    
    let start_time = Instant::now();
    
    for (cmd_name, conn_id, req_id, args) in test_commands {
        let command: Box<dyn Command> = match cmd_name {
            "SET" => {
                match SetCommand::new(args) {
                    Ok(cmd) => Box::new(cmd),
                    Err(e) => {
                        eprintln!("❌ 创建 SET 命令失败: {}", e);
                        continue;
                    }
                }
            }
            "GET" => {
                match GetCommand::new(args) {
                    Ok(cmd) => Box::new(cmd),
                    Err(e) => {
                        eprintln!("❌ 创建 GET 命令失败: {}", e);
                        continue;
                    }
                }
            }
            _ => {
                eprintln!("❌ 不支持的命令: {}", cmd_name);
                continue;
            }
        };
        
        if let Err(e) = storage_processor.process_command_with_response(conn_id, req_id, command) {
            eprintln!("❌ 处理命令失败: {}", e);
        }
        
        // 模拟命令间隔
        thread::sleep(Duration::from_millis(50));
    }
    
    // 5. 高并发测试
    println!("\n⚡ 5. 高并发命令处理测试");
    
    let concurrent_requests = 100;
    let batch_start = Instant::now();
    
    // 使用多个线程模拟并发客户端
    let handles: Vec<_> = (0..4).map(|thread_id| {
        let driver_queue = storage_processor.response_driver.get_queue();
        
        thread::spawn(move || {
            for i in 0..concurrent_requests / 4 {
                let conn_id = (thread_id * 1000 + i) as ConnectionId;
                let req_id = i as RequestId;
                
                // 模拟存储操作完成，直接提交响应
                let response_event = ResponseEvent::new(
                    conn_id,
                    req_id,
                    CommandResult::String(format!("concurrent_value_{}", i)),
                    "CONCURRENT_GET".to_string(),
                    current_timestamp_millis(),
                );
                
                let metadata = redis_rs2::event::EventMetadata::new("response")
                    .with_priority(redis_rs2::event::Priority::Normal)
                    .with_source("concurrent-test");
                
                let event_wrapper = redis_rs2::event::Event::with_metadata(metadata, response_event);
                
                if let Err(_) = driver_queue.enqueue(event_wrapper) {
                    eprintln!("  ❌ 并发提交失败: 队列已满");
                    break;
                }
            }
            
            println!("  ✅ 并发线程 {} 完成处理", thread_id);
        })
    }).collect();
    
    // 等待所有线程完成
    for handle in handles {
        handle.join().unwrap();
    }
    
    let batch_duration = batch_start.elapsed();
    println!("  📊 并发处理完成: {}个请求，耗时: {:?}", concurrent_requests, batch_duration);
    
    // 6. 等待所有响应处理完成
    println!("\n⏱️ 6. 等待响应处理完成");
    
    let mut last_processed = 0;
    for _ in 0..20 { // 最多等待20秒
        let stats = storage_processor.response_driver.get_stats();
        let queue_size = storage_processor.response_driver.queue_size();
        
        if stats.total_responses_processed != last_processed {
            println!(
                "  📈 已处理响应: {} 队列大小: {} 成功率: {:.1}%",
                stats.total_responses_processed,
                queue_size,
                if stats.total_responses_processed > 0 {
                    stats.successful_writebacks as f64 / stats.total_responses_processed as f64 * 100.0
                } else { 0.0 }
            );
            last_processed = stats.total_responses_processed;
        }
        
        if queue_size == 0 && stats.total_responses_processed > 0 {
            break;
        }
        
        thread::sleep(Duration::from_millis(500));
    }
    
    // 7. 打印详细统计
    println!("\n📊 7. 集成统计信息");
    
    println!("\n存储层统计:");
    println!("  {}", storage_processor.get_storage_stats());
    
    println!("\n响应回写统计:");
    storage_processor.response_driver.print_stats();
    
    println!("\n网络层统计:");
    println!("  最大连接ID: {}", network_manager.get_max_connection_id());
    
    // 8. 停止系统
    println!("\n🛑 8. 停止系统");
    if let Err(e) = storage_processor.response_driver.stop() {
        eprintln!("❌ 停止响应驱动器失败: {}", e);
    }
    
    let total_duration = start_time.elapsed();
    println!("\n🎉 集成演示完成！");
    println!("  总运行时间: {:?}", total_duration);
    
    let total_responses = storage_processor.response_driver.get_stats().total_responses_processed;
    if total_duration.as_secs_f64() > 0.0 {
        println!("  平均吞吐量: {:.0} 响应/秒", 
            total_responses as f64 / total_duration.as_secs_f64());
    }
    
    println!("\n💡 集成要点:");
    println!("  ✅ 存储层命令执行与响应回写解耦");
    println!("  ✅ 异步响应处理不阻塞存储操作");
    println!("  ✅ 支持高并发请求处理");
    println!("  ✅ 完整的统计和监控能力");
    println!("  ✅ 网络层回调机制就绪");
    println!("  🚀 为完整 Redis 协议支持打下基础");
}

/// 获取当前时间戳（毫秒）
fn current_timestamp_millis() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64
}